from pyspark import SparkConf
from pyspark import SparkContext

'''
需求:统计出每一个省份广告被点击次数的TOP3
(逆推思路)
1、统计出每个省份中每个广告的点击次数的降序排序(降序排序)
2、统计出每个省份中每个广告的点击次数(求和)
3、基于每个省份和每个广告来分组(group)
4、把省份和广告同时都作为键
(key有两部分组成)
5、把数据切割
'''


def func_map(line):
    arr = line.split(' ')
    return (arr[1], arr[4]), 1  # 把省份和广告封装成一个元祖。该元祖就key


def func_mapValues(iter):  # iter是一个迭代器:代表一个省份中所有广告及点击次数
    return sorted(iter, key=lambda t: t[1], reverse=True)[:3]  # 取TOP3


if __name__ == '__main__':
    conf = SparkConf().setMaster("local[1]").setAppName("spark01")
    sc = SparkContext(conf=conf)

    # 读取数据
    line_rdd = sc.textFile('../data/data.log')
    # 切割数据
    # 第二个map算子: 把原来的key( 省份+ 广告) 拆分成 (省份) ，原理的(点击次数) 改成 (广告，点击次数)
    # groupByKey 之后:key,value (list(该省份下所有的广告及点击次数))
    # mapValues :必须是基于K,V的RDD, 把原来RDD中的每条数据的值做转换
    result = line_rdd.map(func_map) \
        .reduceByKey(lambda x, y: x + y) \
        .map(lambda t: (t[0][0], (t[0][1], t[1]))) \
        .groupByKey() \
        .mapValues(func_mapValues)

    result.foreach(print)
